package com.wudl.flink.examples;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName : FlinkSqlMysqlToMySql
 * @Description : Flink sql-mysql
 * @Author :wudl
 * @Date: 2021-08-24 23:28
 */

public class FlinkSqlMysqlToElasticsearch02 {
    public static void main(String[] args) {

        String driverClass = "com.mysql.jdbc.Driver";

        String dbUrl = "jdbc:mysql://192.168.1.180:3306/MyFlink";
        String userNmae = "root";
        String passWord = "123456";
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
        StreamTableEnvironment tableEvn = StreamTableEnvironment .create(env,settings);
        //1. 指定方言
        tableEvn.getConfig().setSqlDialect(SqlDialect.DEFAULT);
        String flink_sink_table = "myEs";

        TableResult inputTable = tableEvn.executeSql("CREATE TABLE  esTable (" +
                "name STRING ," +
                "sex STRING " +
                ") " +
                "WITH (" +
                "'connector' = 'elasticsearch-7'," +
                "'hosts' = 'http://192.168.1.162:9200'," +
                "'index' = 'wudl_test'"+
                " )");
      TableResult outPutTable = tableEvn.executeSql("CREATE TABLE  sourceMySqlTable (" +
              "id int ," +
              "name STRING ," +
              "sex STRING " +
              ") " +
                "WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = '" + dbUrl + "'," +
                "'table-name' = '"+flink_sink_table+"'," +
                " 'username' = '" + userNmae + "'," +
                " 'password' = '" + passWord + "'" +
                " )");

        String sql = " select id,name,sex  from sourceMySqlTable";
        Table ResultTable = tableEvn.sqlQuery(sql);

        tableEvn.executeSql("insert into esTable select name,sex  from "+ResultTable);



    }
}
